package ds_industry_2025.ds.ds01.T4

import org.apache.spark.sql.functions._
import org.apache.spark.sql.{DataFrame, SparkSession}

import java.util.Properties

/*
    1、根据Hive的dwd库中相关表或MySQL中shtd_store中相关表（order_detail、sku_info），计算出与用户id为6708的用户所购买相同
    商品种类最多的前10位用户（只考虑他俩购买过多少种相同的商品，不考虑相同的商品买了多少次），将10位用户id进行输出，输出格式如下，
    将结果截图粘贴至客户端桌面【Release\任务C提交结果.docx】中对应的任务序号下；
结果格式如下：
-------------------相同种类前10的id结果展示为：--------------------
1,2,901,4,5,21,32,91,14,52
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    // todo 定义读取mysql表数据的方法
    def read_mysql(table_name:String):DataFrame={
      spark.read.jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false",table_name,conn)
    }

    //  todo 读取mysql的数据
    val order_info=read_mysql("order_info")
    val detail_info=read_mysql("order_detail")
    val user_info=read_mysql("user_info")
    val sku_info=read_mysql("sku_info")

    //  todo 剔除订单表和订单详细表里面user_id,sku_id不存在于维度表里面的数据
    val skus=sku_info.select("id").distinct().withColumnRenamed("id","id_1")
    val users=user_info.select("id").distinct().withColumnRenamed("id","id_1")

    val detail = detail_info.join(skus, skus("id_1") === detail_info("sku_id"),"inner")
    val order=order_info.join(users,users("id_1")===order_info("user_id"),"inner")

    // todo 将所有用户购买过商品的数据拿到
    val data = detail.join(order, detail("order_id") === order("id"))
      .select("user_id", "sku_id")
      .distinct()

    //  todo 选出6708用户所购买的商品
    val user6780_skus:Array[Int]= data.filter(col("user_id") === 6708).select("sku_id")
      .map(r => r(0).toString.toInt).collect()


    //  todo 判断用户购买的商品是否在6780用户购买的商品里面，是的话cos值为1，然后sum(cos)的值得到购买种类最多的用户
    val result:String=data.withColumn(
      "cos",
      when(
        col("sku_id").isin(user6780_skus: _*),1.0
      ).otherwise(0.0)
    )
      .groupBy("user_id")
      .agg(sum("cos").as("same"))
      .filter(col("user_id") !== 6780)
      .orderBy(desc("same"),asc("user_id"))
      .limit(10)
      .select("user_id")
      .map(r => r(0).toString)
      .collect()
      .mkString(",")

    println(result)


    spark.close()

  }

}
